import java.io.BufferedWriter;
import java.io.FileWriter;
import java.io.IOException;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.SocketException;
import java.net.SocketTimeoutException;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Iterator;
import java.util.Scanner;
import java.util.Random;
import java.util.StringTokenizer;

/**
 * The Paxos Peer that assumes multiple roles
 */

public class PaxosPeer 
{

	int pid; /* Process PID */
//	PaxosMessage highestAcceptedProposal; /* Last proposal it accepted */
//	PaxosMessage highestPrepareResponded; /* Most recent proposal it heard */
	int port; /* UDP port to run on */
	String remoteHost;
	String logFile;
	float dropProbability;
	DatagramSocket pSocket;
	ArrayList<Integer > upSet;
	int currentRole;
	int currentLeader = 0;
	boolean watingForLeader = false;
	int timeoutBound;
	int currentInstanceId;
	int retriesBound = 0;
	// leader DS
	ArrayList<Integer> instancePrepareProposalNum = new ArrayList<Integer>();
	ArrayList<String> instanceProposalVal = new ArrayList<String>();

	//acceptor DS
	ArrayList<Integer> instanceHighestAcceptProposalNum = new ArrayList<Integer>();
	ArrayList<String> instanceHighestAcceptProposalVal = new ArrayList<String>();
	
	ArrayList<Integer> instanceHighestPrepareProposalNum = new ArrayList<Integer>();

	int clientAccount;
	int tentativeBal = 0;
	
	ArrayList<String> instanceResponseList = new ArrayList<String>();
	
	ArrayList<String> instanceLearnList = new ArrayList<String>();
	
	ArrayList<Integer> instanceNumPromiseReceived = new ArrayList<Integer>();
	
	ArrayList<Integer> instanceNumChosenReceived = new ArrayList<Integer>();
	
	ArrayList<String> allInboundPromises = new ArrayList<String>();
	
	ArrayList<String> allLeaderMesgs = new ArrayList<String>();
	
	InetSocketAddress clientSocket = null;
	
	byte[] buf = new byte[PaxosConsts.MESG_SIZE];
	
	public PaxosPeer() {
		// TODO Auto-generated constructor stub
		
		dropProbability = 0;
		currentInstanceId=0;
		clientAccount = 0;
	}
	
	public PaxosPeer(int pid, int port, float dropProbability, int currentRole, String remoteHost, ArrayList<Integer> upSet, int timeoutBound, int retriesBound)
	{
		this();
		this.pid = pid;
		this.port = port;
		this.dropProbability = dropProbability;
		this.upSet = upSet;
		this.currentRole = currentRole;
		this.remoteHost = remoteHost;
		this.timeoutBound = timeoutBound; 
		this.retriesBound = retriesBound;
		logFile = currentRole+""+pid+".log";
		try
		{
			pSocket = new DatagramSocket(port);
		}
		catch(SocketException se)
		{
			System.err.println(se.getMessage());
		}
		System.out.println("PID: "+pid+" - I am a new process running at port:"+this.port+" with drop probability: "+this.dropProbability);
		
		for (int i=0;i<PaxosConsts.INSTANCE_SIZE;i++)
		{
			instancePrepareProposalNum.add(PaxosConsts.NO_ELEMENT);
			instanceProposalVal.add(PaxosConsts.NO_ELE_STR);
			instanceHighestAcceptProposalNum.add(PaxosConsts.NO_ELEMENT);
			instanceHighestAcceptProposalVal.add(PaxosConsts.NO_ELE_STR);
			instanceHighestPrepareProposalNum.add(PaxosConsts.NO_ELEMENT);
			instanceResponseList.add(PaxosConsts.NO_ELE_STR);
			instanceLearnList.add(PaxosConsts.NO_ELE_STR);
			instanceNumPromiseReceived.add(PaxosConsts.NO_ELEMENT);
			instanceNumChosenReceived.add(PaxosConsts.NO_ELEMENT);
		}
		
	}
	
	public boolean dropPacket()
	{
		// Drop packet based on peer probability
		// Say drop probability is .3
		// Generate a random number between [0,9). If the random number is less than 3 return true.
		// So indirectly we generate a number with probability of 33%. 
		Random rand = new Random();
		if (rand.nextInt(PaxosConsts.PROBABILTY_BOUND) < (dropProbability*PaxosConsts.PROBABILTY_BOUND))
			return true;
		else
			return false;
	}
	
	
	public boolean duplicateLeaderMesg(DatagramPacket inboundMesg)
	{
		boolean duplicate = false;
		String paxosStrMesg  = new String(inboundMesg.getData());
		if (allLeaderMesgs.indexOf(paxosStrMesg) != -1)
		{
			duplicate = true;
		}
		else
		{
			allLeaderMesgs.add(paxosStrMesg);
		}
		return duplicate;
	}
	
	/*
	 * Roles that can be played
	 */

	public int playAcceptor()
	{
		int protocolState = PaxosConsts.CONTINUE;
		DatagramPacket outboundPacket = null;
		int acceptorRetry = 0;
		try
		{
			pSocket.setSoTimeout(timeoutBound);
		}
		catch (SocketException e) {
			System.err.println(e.getMessage());
		}
		while(protocolState == PaxosConsts.CONTINUE)
		{
			DatagramPacket inboundMesg = new DatagramPacket(buf,buf.length);
			try
			{
				Arrays.fill(inboundMesg.getData(), 0,PaxosConsts.MESG_SIZE,(byte)0);
				pSocket.receive(inboundMesg);
				PaxosMessage inboundPaxosMesg = new PaxosMessage(inboundMesg.getData());
				acceptorRetry=0;
				// Drop messages based on the drop probability
				if(dropPacket())
				{
					System.err.println("PID: "+pid+" - Dropping packet <"+inboundPaxosMesg.msgType+" for instance : "+inboundPaxosMesg.instanceId+"> received from PID: "+inboundPaxosMesg.senderId);
				}
				else
				{				
					if(duplicateLeaderMesg(inboundMesg))
					{
						continue;
					}
					
					
					if (inboundPaxosMesg.msgType.equals(PaxosConsts.PREPARE)) 
					{
						System.out.println("PID: "+pid+" - received < "+inboundPaxosMesg.msgType+" for instance: "+inboundPaxosMesg.instanceId+" with proposal number: "+inboundPaxosMesg.proposalNum+"> from PID:"+inboundPaxosMesg.senderId);
						outboundPacket = processPrepare(inboundPaxosMesg);
						watingForLeader = true;

					}
					else if (inboundPaxosMesg.msgType.equals(PaxosConsts.ACCEPT))
					{
						System.out.println("PID: "+pid+" - received < "+inboundPaxosMesg.msgType+" for instance: "+inboundPaxosMesg.instanceId+" with value:"+inboundPaxosMesg.proposalValue+"> from PID:"+inboundPaxosMesg.senderId);
						outboundPacket = processAccept(inboundPaxosMesg);
						watingForLeader = true;
					}
					else if (inboundPaxosMesg.msgType.equals(PaxosConsts.LEARN))
					{
						System.out.println("PID: "+pid+" - received < "+inboundPaxosMesg.msgType+" for instance: "+inboundPaxosMesg.instanceId+" with value:"+inboundPaxosMesg.respToClient+" > from PID:"+inboundPaxosMesg.senderId);
						System.out.println("DEBUG incoming learn: "+new String(inboundMesg.getData()));
						protocolState = processLearn(inboundPaxosMesg);
					}
					else if (inboundPaxosMesg.msgType.equals(PaxosConsts.CLIENT_REQ))
					{
						System.out.println("PID: "+pid+" - received < "+inboundPaxosMesg.msgType+"> from CID:"+inboundPaxosMesg.senderId+". Ignoring packet.");
					}
				}
			}
			catch (SocketTimeoutException e) 
			{
				System.err.println(e.getMessage());
				acceptorRetry = acceptorRetry+1;
				System.out.println("PID: "+pid+" Timedout on PID: "+currentLeader+". Retry number is: "+acceptorRetry);
				if (acceptorRetry < retriesBound)
				{
					if (watingForLeader)
					{
						try
						{
							PaxosMessage resentPaxosMesg = new PaxosMessage(outboundPacket.getData());
							System.out.println("PID: "+pid+" - resending < "+resentPaxosMesg.msgType+" for instance: "+resentPaxosMesg.instanceId+" having value: "+resentPaxosMesg.proposalValue+"> to PID:"+currentLeader);
							pSocket.send(outboundPacket);
						}
						catch (IOException io) {
							System.err.println(io.getMessage());
						}
					}
				}
				else
				{
					watingForLeader = false;
					upSet.set(currentLeader, PaxosConsts.DEAD_PROCESS_PORT);
					System.out.println("PID: "+pid+" - LEADER with PID: "+currentLeader+" HAS CRASHED!");
					protocolState = PaxosConsts.LEADER_DIED;
				}
			}
			catch (IOException e) 
			{
				// TODO: handle exception
			}
		}
		
		
		return protocolState;
	}
	
	public int processLearn(PaxosMessage inboundPaxosMessage) {
		// TODO Auto-generated method stub
		// WRITE TO DISK CLIENT RESPONSE
		int protocolState = PaxosConsts.CONTINUE;
		instanceResponseList.set(inboundPaxosMessage.instanceId, inboundPaxosMessage.respToClient);
		instanceLearnList.set(inboundPaxosMessage.instanceId, inboundPaxosMessage.respToClient);
		if (inboundPaxosMessage.proposalValue.indexOf(PaxosConsts.BALANCE) != -1)
		{
			// not a state changing thing
		}
		if (inboundPaxosMessage.proposalValue.indexOf(PaxosConsts.WITHDRAW) != -1)
		{
			StringTokenizer withdrawTokens = new StringTokenizer(inboundPaxosMessage.proposalValue);
			withdrawTokens.nextToken(); 
			clientAccount = clientAccount-Integer.parseInt(withdrawTokens.nextToken());
			
		}
		if (inboundPaxosMessage.proposalValue.indexOf(PaxosConsts.DEPOSIT) != -1)
		{
			StringTokenizer depositTokens = new StringTokenizer(inboundPaxosMessage.proposalValue);
			depositTokens.nextToken(); 
			clientAccount = clientAccount +Integer.parseInt(depositTokens.nextToken());
			
		}
		if (inboundPaxosMessage.proposalValue.indexOf(PaxosConsts.EXIT) != -1)
		{
			protocolState = PaxosConsts.SESSION_DONE;
		}
		writeLog("Final client account state: "+clientAccount);
		watingForLeader = false;
		return protocolState;
		
	}

	public String tentativeClientAccount(String proposalValGot, int clientID) 
	{
		// TODO Auto-generated method stub
		String returnMesgToLeader = "";
		if (proposalValGot.indexOf(PaxosConsts.BALANCE) != -1)
		{
			returnMesgToLeader = PaxosConsts.BALANCE_RESPONSE+":"+tentativeBal;
		}
		if (proposalValGot.indexOf(PaxosConsts.WITHDRAW) != -1)
		{
			StringTokenizer withdrawTokens = new StringTokenizer(proposalValGot);
			withdrawTokens.nextToken(); 
			tentativeBal = tentativeBal-Integer.parseInt(withdrawTokens.nextToken());
			returnMesgToLeader = PaxosConsts.WITHDRAW_RESPONSE+":"+tentativeBal;
		}
		if (proposalValGot.indexOf(PaxosConsts.DEPOSIT) != -1)
		{
			StringTokenizer depositTokens = new StringTokenizer(proposalValGot);
			depositTokens.nextToken(); 
			tentativeBal = tentativeBal+Integer.parseInt(depositTokens.nextToken());
			returnMesgToLeader = PaxosConsts.DEPOSIT_RESPONSE+":"+tentativeBal;
		}
		if (proposalValGot.indexOf(PaxosConsts.EXIT) != -1)
		{
			returnMesgToLeader = PaxosConsts.EXIT+":"+tentativeBal;
		}
		return returnMesgToLeader;
	}
	public DatagramPacket processPrepare(PaxosMessage inboundPrepareMsg) 
	{
		
		int proposalNumGot = inboundPrepareMsg.proposalNum;
		int instanceIDGot = inboundPrepareMsg.instanceId;
		int clientID = inboundPrepareMsg.clientId;
		byte promise[] = null;
		if (!instanceLearnList.get(instanceIDGot).equals(PaxosConsts.NO_ELE_STR))
		{
			System.out.println("PID: "+pid+" received PREPARE request for instance: "+inboundPrepareMsg.instanceId+" who's value: "+instanceLearnList.get(instanceIDGot)+" has already been LEARNT");
			promise = (new PaxosMessage()).buildPaxosMsg(PaxosConsts.ALREADY_LEARNED,pid,instanceIDGot,proposalNumGot,
					PaxosConsts.NO_ELE_STR,null,instanceHighestAcceptProposalNum.get(instanceIDGot),
					instanceHighestAcceptProposalVal.get(instanceIDGot),instanceLearnList.get(instanceIDGot),clientID);
			System.out.println("PID: "+pid+" - sending < "+PaxosConsts.ALREADY_LEARNED+" for instance: "+instanceIDGot+" to PID: "+currentLeader);
		}
		else if (!instanceResponseList.get(instanceIDGot).equals(PaxosConsts.NO_ELE_STR) 
				&& instanceLearnList.get(instanceIDGot).equals(PaxosConsts.NO_ELE_STR))
		{
			System.out.println("PID: "+pid+" received PREPARE request for instance: "+inboundPrepareMsg.instanceId+" who's value: "+instanceResponseList.get(instanceIDGot)+" has already been CHOSEN");
				promise = (new PaxosMessage()).buildPaxosMsg(PaxosConsts.ALREADY_PROCESSED,pid,instanceIDGot,proposalNumGot,
						PaxosConsts.NO_ELE_STR,null,instanceHighestAcceptProposalNum.get(instanceIDGot),
						instanceHighestAcceptProposalVal.get(instanceIDGot),instanceResponseList.get(instanceIDGot),clientID);
				System.out.println("PID: "+pid+" - sending < "+PaxosConsts.ALREADY_PROCESSED+" for instance: "+instanceIDGot+" to PID: "+currentLeader);
		}
		else
		if (proposalNumGot <= instanceHighestPrepareProposalNum.get(instanceIDGot))
		{
			// TODO nothing
			System.out.println("PID: "+pid+" received PREPARE request for instance: "+inboundPrepareMsg.instanceId+" who's proposal number is NOT GREATER than " 
					+"the highest proposal number: "+instanceHighestPrepareProposalNum.get(instanceIDGot)+" seen in this instance");
			promise = (new PaxosMessage()).buildPaxosMsg(PaxosConsts.PREPARE_REJECT,pid,instanceIDGot,instanceHighestPrepareProposalNum.get(instanceIDGot),
					PaxosConsts.NO_ELE_STR,
					null,instanceHighestAcceptProposalNum.get(instanceIDGot),instanceHighestAcceptProposalVal.get(instanceIDGot),PaxosConsts.NO_ELE_STR,
					clientID);
			System.out.println("PID: "+pid+" - sending < "+PaxosConsts.PREPARE_REJECT+" for instance: "+instanceIDGot+" to PID: "+currentLeader+" > ");
			

		}
		else if (proposalNumGot > instanceHighestPrepareProposalNum.get(instanceIDGot))
		{
			instanceHighestPrepareProposalNum.set(instanceIDGot,proposalNumGot);
			promise = (new PaxosMessage()).buildPaxosMsg(PaxosConsts.PROMISE,pid,instanceIDGot,proposalNumGot,PaxosConsts.NO_ELE_STR,
					null,instanceHighestAcceptProposalNum.get(instanceIDGot),instanceHighestAcceptProposalVal.get(instanceIDGot),PaxosConsts.NO_ELE_STR,
					clientID);
			System.out.println("PID: "+pid+" - sending < "+PaxosConsts.PROMISE+" for instance: "+instanceIDGot+" to PID: "+currentLeader+" > ");
			writeLog(PaxosConsts.PROMISE+" instance: "+instanceIDGot+" to PID: "+currentLeader);
		}
		DatagramPacket promisePacket = null; 
			
		try
		{
			promisePacket = new DatagramPacket(promise, promise.length, InetAddress.getByName(remoteHost),upSet.get(currentLeader));
			Scanner yesOrNo = new Scanner(System.in);
			yesOrNo.nextLine();
			pSocket.send(promisePacket);
		}
		catch (UnknownHostException he) 
		{
			System.err.println(he.getMessage());
		} catch (IOException e) {
			// TODO Auto-generated catch block
			System.err.println(e.getMessage());
		}
		return promisePacket;
		
	}

	public DatagramPacket processAccept(PaxosMessage inboundAcceptMsg) 
	{
		int proposalNumGot = inboundAcceptMsg.proposalNum;
		int instanceIDGot = inboundAcceptMsg.instanceId;
		String proposalValGot = inboundAcceptMsg.proposalValue;
		int clientID = inboundAcceptMsg.clientId;
		DatagramPacket chosenPacket = null;
		if (proposalNumGot >= instanceHighestPrepareProposalNum.get(instanceIDGot))
		{
			// update current instance highs
			instanceHighestAcceptProposalNum.set(instanceIDGot,proposalNumGot);
			instanceHighestAcceptProposalVal.set(instanceIDGot,proposalValGot);
			String respToClient = tentativeClientAccount(proposalValGot,clientID);
			instanceResponseList.set(instanceIDGot, respToClient);
			byte chosen[] = (new PaxosMessage()).buildPaxosMsg(PaxosConsts.CHOSEN,pid,instanceIDGot,proposalNumGot,proposalValGot,
					null,PaxosConsts.NO_ELE_INT,PaxosConsts.NO_ELE_STR,respToClient,clientID);
			
			try
			{
				chosenPacket = new DatagramPacket(chosen, chosen.length, InetAddress.getByName(remoteHost),upSet.get(currentLeader));
				System.out.println("PID: "+pid+" - sending < "+PaxosConsts.CHOSEN+" for instance: "+instanceIDGot+" having value: "+proposalValGot+"> to PID: "+currentLeader);
				writeLog(PaxosConsts.CHOSEN+" instance: "+instanceIDGot+" value: "+proposalValGot+" to PID: "+currentLeader);
				Scanner yesOrNo = new Scanner(System.in);
				yesOrNo.nextLine();
				pSocket.send(chosenPacket);
			}
			catch (UnknownHostException he) 
			{
				System.err.println(he.getMessage());
			} catch (IOException e) {
				// TODO Auto-generated catch block
				System.err.println(e.getMessage());
			}
		}
		return chosenPacket;
	}

	

	public int processChosen(PaxosMessage chosenMesg, String learned)
	{
		int heardExit = PaxosConsts.CONTINUE;
		instanceResponseList.set(chosenMesg.instanceId,chosenMesg.respToClient);
		instanceLearnList.set(chosenMesg.instanceId,chosenMesg.respToClient);
		if (!(instanceResponseList.indexOf(PaxosConsts.NO_ELE_STR) < chosenMesg.instanceId))
		{
			if (learned.equals(PaxosConsts.ALREADY_PROCESSED))
			{
				byte [] learnMsg  = (new PaxosMessage()).buildPaxosMsg(PaxosConsts.LEARN, pid, 
						chosenMesg.instanceId, chosenMesg.proposalNum, chosenMesg.proposalValue, null, chosenMesg.highAcceptProposalNum, 
						chosenMesg.highAcceptProposalValue,chosenMesg.respToClient,chosenMesg.clientId);
				
				for (int j=0;j<upSet.size();j++)
				{
		 			try
					{
		 				if (upSet.get(j) != PaxosConsts.DEAD_PROCESS_PORT && j!=pid)
		 				{
		 					for (int i=0;i<PaxosConsts.LEADER_RESEND_BOUND;i++)
		 					{
		 						pSocket.send(new DatagramPacket(learnMsg,learnMsg.length,InetAddress.getByName(remoteHost),upSet.get(j)));
		 					}
		 					System.out.println("PID: "+pid+" - sending < "+PaxosConsts.LEARN+" for instance: "+chosenMesg.instanceId+" having value: "+chosenMesg.respToClient+"> to PID: "+j);
		 					writeLog(PaxosConsts.LEARN+" instance: "+chosenMesg.instanceId+" value: "+chosenMesg.respToClient.trim().toString()+" to PID: "+j);
		 				}
		 				
					}
					catch(IOException io)
					{
						System.err.println(io.getMessage());
					}
				}
			}
			heardExit = sendToClient(chosenMesg.instanceId,chosenMesg.respToClient);
			int i = chosenMesg.instanceId+1;
			// other pending
			while(!(instanceResponseList.get(i).equals(PaxosConsts.NO_ELE_STR)))
			{
				heardExit = sendToClient(i,instanceResponseList.get(i));
				i++;
				byte [] restlearnMsg  = (new PaxosMessage()).buildPaxosMsg(PaxosConsts.LEARN, pid, 
						i, chosenMesg.proposalNum, chosenMesg.proposalValue, null, chosenMesg.highAcceptProposalNum, 
						chosenMesg.highAcceptProposalValue,instanceResponseList.get(i),chosenMesg.clientId);
				
				for (int j=0;j<upSet.size();j++)
				{
		 			try
					{
		 				if (upSet.get(j) != PaxosConsts.DEAD_PROCESS_PORT && j!=pid)
		 				{
		 					for (int k=0;k<PaxosConsts.LEADER_RESEND_BOUND;k++)
		 					{
		 						pSocket.send(new DatagramPacket(restlearnMsg,restlearnMsg.length,InetAddress.getByName(remoteHost),upSet.get(j)));
		 					}
		 					System.out.println("PID: "+pid+" - sending < "+PaxosConsts.LEARN+" for instance: "+chosenMesg.instanceId+" having value: "+chosenMesg.respToClient+"> to PID: "+j);
		 				}
					}
					catch(IOException io)
					{
						System.err.println(io.getMessage());
					}
				}
			}
		}
		return heardExit;
	}
	
	public int sendToClient(int instanceId, String response) 
	{
		instanceResponseList.set(instanceId,response);
		StringTokenizer respAndValue = new StringTokenizer(response,":");
		String resp = respAndValue.nextToken();
		String value = respAndValue.nextToken();
		String finalClientResp = "";

		if (resp.equals(PaxosConsts.WITHDRAW_RESPONSE) || resp.equals(PaxosConsts.DEPOSIT_RESPONSE) || resp.equals(PaxosConsts.EXIT))
		{
			finalClientResp = resp;
		}
		else
		{
			finalClientResp = resp+value;	
		}
		byte [] respClientMsg  = (new PaxosMessage()).buildPaxosMsg(PaxosConsts.CLIENT_RSP, pid, 
				instanceId, PaxosConsts.NO_ELEMENT, PaxosConsts.NO_ELE_STR, null, PaxosConsts.NO_ELEMENT, 
				PaxosConsts.NO_ELE_STR,finalClientResp,PaxosConsts.NO_ELEMENT);
		try
		{
			System.out.println("PID: "+pid+" - sending < "+PaxosConsts.CLIENT_RSP+" for instance: "+instanceId+" having requested operation: "+instanceProposalVal.get(instanceId)+
					" with response having value: "+finalClientResp+"> to Client");
			pSocket.send(new DatagramPacket(respClientMsg,respClientMsg.length,clientSocket));
		}
		catch(IOException io)
		{
			System.err.println(io.getMessage());
		}
		if(response.indexOf(PaxosConsts.EXIT) != -1)
		{
			return PaxosConsts.SESSION_DONE;
		}
		return PaxosConsts.CONTINUE;
	}
	
	public void processPromise(ArrayList<String> allInboundPromises,int promiseOfInstance)
	{
	
	    int highestAcceptProposalNum=PaxosConsts.NO_ELE_INT;
	    String highestAcceptProposalVal=PaxosConsts.NO_ELE_STR;
	    int proposalNum = PaxosConsts.NO_ELEMENT;
	    String newProposalVal = PaxosConsts.NO_ELE_STR;
	    int clientID = PaxosConsts.NO_ELEMENT;
	    for (int i=0;i<allInboundPromises.size();i++)
	    {
	    	PaxosMessage promiseMsg= new PaxosMessage(allInboundPromises.get(i));
	    	if (promiseMsg.instanceId == promiseOfInstance)
	    	{
		    	clientID = promiseMsg.clientId;
		    	if (promiseMsg.highAcceptProposalNum > highestAcceptProposalNum)
		    	{
		    		highestAcceptProposalNum = promiseMsg.highAcceptProposalNum;
		    		highestAcceptProposalVal =promiseMsg.highAcceptProposalValue;
		    		proposalNum = promiseMsg.proposalNum;
		    	}
	    	}
	    }
	
	    if (highestAcceptProposalVal.equals(PaxosConsts.NO_ELE_STR))
	    {
	    	newProposalVal = instanceProposalVal.get(promiseOfInstance);
	    }
	    else
	    {
	    	newProposalVal = highestAcceptProposalVal;
	    }
		byte [] acceptMsg = (new PaxosMessage()).buildPaxosMsg(PaxosConsts.ACCEPT, pid, promiseOfInstance, proposalNum, newProposalVal, null, 
				PaxosConsts.NO_ELE_INT, PaxosConsts.NO_ELE_STR,PaxosConsts.NO_ELE_STR,clientID);
		for (int j=0;j<upSet.size();j++)
		{
			if (upSet.get(j) != PaxosConsts.DEAD_PROCESS_PORT && j!=pid)
			{
				try
				{
					if (upSet.get(j) != PaxosConsts.DEAD_PROCESS_PORT && j!=pid)
	 				{
						System.out.println("PID: "+pid+" - sending < "+PaxosConsts.ACCEPT+" for instance: "+promiseOfInstance+" having value: "+newProposalVal+"> to PID:"+j);
						writeLog(  PaxosConsts.ACCEPT+" for instance: "+promiseOfInstance+" value: "+newProposalVal.trim().toString()+" to PID:"+j);
	 					for (int i=0;i<PaxosConsts.LEADER_RESEND_BOUND;i++)
	 					{
	 						pSocket.send(new DatagramPacket(acceptMsg,acceptMsg.length,InetAddress.getByName(remoteHost),upSet.get(j)));
	 					}
	 				}
				}
				catch(IOException io)
				{
					System.err.println(io.getMessage());
				}
			}
		}
	
	}
	


	public int processClientReq(PaxosMessage inboundClientMesg)
	{
		int protocolState = PaxosConsts.CONTINUE;
	    String cmd = "";
	    int numCmds = inboundClientMesg.cmdParamList.size();
	    int loopCmd=0;
	    while (loopCmd< numCmds) 
	    {
			cmd = inboundClientMesg.cmdParamList.get(loopCmd).trim();
			if (cmd.equals(""))
			{
				cmd = PaxosConsts.NO_ELE_STR;
			}
			if (cmd.equals(PaxosConsts.NO_ELE_STR))
			{
				loopCmd++;
				continue;
			}
			if (!instanceLearnList.get(loopCmd).equals(PaxosConsts.NO_ELE_STR))
			{
				System.out.println("PID: "+pid+" received client request: "+cmd+" for instance: "+loopCmd+" who's value: "+instanceLearnList.get(loopCmd)+" has already been LEARNT");
				inboundClientMesg.respToClient = instanceLearnList.get(loopCmd);
				inboundClientMesg.instanceId = loopCmd;
				protocolState = processChosen(inboundClientMesg, PaxosConsts.ALREADY_LEARNED);
				loopCmd++;
			}
			else if (!instanceResponseList.get(loopCmd).equals(PaxosConsts.NO_ELE_STR) 
					&& instanceLearnList.get(loopCmd).equals(PaxosConsts.NO_ELE_STR))
			{
				System.out.println("PID: "+pid+" received client request: "+cmd+" for instance: "+loopCmd+" who's value: "+instanceResponseList.get(loopCmd)+" has already been CHOSEN");
				inboundClientMesg.instanceId = loopCmd;
				inboundClientMesg.respToClient = instanceResponseList.get(loopCmd);
				protocolState = processChosen(inboundClientMesg, PaxosConsts.ALREADY_PROCESSED);
				loopCmd++;
			}
			else if (protocolState == PaxosConsts.CONTINUE)
			{
				int clientID = inboundClientMesg.senderId;
				
				int currPreparePropNum = instancePrepareProposalNum.get(loopCmd);
				instancePrepareProposalNum.set(loopCmd, ++currPreparePropNum);
				instanceProposalVal.set(loopCmd, cmd);
				
				byte [] prepareMsg  = (new PaxosMessage()).buildPaxosMsg(PaxosConsts.PREPARE, pid, 
						loopCmd, currPreparePropNum, cmd, null, PaxosConsts.NO_ELE_INT, 
						PaxosConsts.NO_ELE_STR,PaxosConsts.NO_ELE_STR,clientID);
				
	 			for (int j=0;j<upSet.size();j++)
	 			{
		 			try
					{
		 				if (upSet.get(j) != PaxosConsts.DEAD_PROCESS_PORT && j!=pid)
		 				{
		 					for (int i=0;i<PaxosConsts.LEADER_RESEND_BOUND;i++)
		 					{
		 						pSocket.send(new DatagramPacket(prepareMsg,prepareMsg.length,InetAddress.getByName(remoteHost),upSet.get(j)));
		 					}
		 					System.out.println("PID: "+pid+" - sending < "+PaxosConsts.PREPARE+" for instance: "+loopCmd+" with proposal number:"+currPreparePropNum+" > to PID:"+j);
		 					writeLog(PaxosConsts.PREPARE+" instance: "+loopCmd+" Proposal number: "+currPreparePropNum+"  to PID: "+j);
		 				}
					}
					catch(IOException io)
					{
						System.err.println(io.getMessage());
					}
	 			}
	 			loopCmd++;
			}
	    }
	    return protocolState;
	}
	
	
	public void processResendPrepare(PaxosMessage inboundPaxosMesg)
	{
		byte [] resendprepareMsg  = (new PaxosMessage()).buildPaxosMsg(PaxosConsts.PREPARE, pid, 
				inboundPaxosMesg.instanceId, inboundPaxosMesg.proposalNum+1, PaxosConsts.NO_ELE_STR, null, PaxosConsts.NO_ELE_INT, 
				PaxosConsts.NO_ELE_STR,PaxosConsts.NO_ELE_STR,inboundPaxosMesg.clientId);
		instancePrepareProposalNum.set(inboundPaxosMesg.instanceId,inboundPaxosMesg.proposalNum+1);
		instanceNumPromiseReceived.set(inboundPaxosMesg.instanceId, 0);
		
		Iterator<String> it = allInboundPromises.iterator();
		while (it.hasNext()) {
			String promiseMsgGot = (String) it.next();
			int instId = (new PaxosMessage(promiseMsgGot)).instanceId;
			if (instId == inboundPaxosMesg.instanceId)
			{
				it.remove();
			}
			
		}
		for (int j=0;j<upSet.size();j++)
		{
			try
			{
				if (upSet.get(j) != PaxosConsts.DEAD_PROCESS_PORT && j!=pid)
				{
					for (int i=0;i<PaxosConsts.LEADER_RESEND_BOUND;i++)
					{
						pSocket.send(new DatagramPacket(resendprepareMsg,resendprepareMsg.length,InetAddress.getByName(remoteHost),upSet.get(j)));
					}
					System.out.println("PID: "+pid+" - resending < "+PaxosConsts.PREPARE+" for instance: "+inboundPaxosMesg.instanceId+" having proposal number: "+instancePrepareProposalNum.get(inboundPaxosMesg.instanceId)+"> to PID:"+j);
				}
			}
			catch(IOException io)
			{
				System.err.println(io.getMessage());
			}
		}
	}
	
	public int playLeader()
	{
		int protocolState = PaxosConsts.CONTINUE;		// TODO: should i be dropping client requests
		DatagramPacket inboundMesg = new DatagramPacket(buf,buf.length);
		try
		{
			pSocket.setSoTimeout(PaxosConsts.NO_ELEMENT);
		}
		catch (SocketException e) {
			System.err.println(e.getMessage());
		}

		while(protocolState == PaxosConsts.CONTINUE)
		{
			try
			{
				Arrays.fill(inboundMesg.getData(), 0,PaxosConsts.MESG_SIZE,(byte)0);
				System.out.println("PID: "+pid+" I am the Leader, listening on port : "+port);
				pSocket.receive(inboundMesg);
				PaxosMessage inboundPaxosMesg = new PaxosMessage(inboundMesg.getData());
				if(dropPacket())
 		    	{
					System.err.println("PID: "+pid+" - dropping packet received from PID: "+inboundPaxosMesg.senderId);
					continue;
 		    	}
				
				if (inboundPaxosMesg.msgType.equals(PaxosConsts.CLIENT_REQ))
				{
					clientSocket = new InetSocketAddress(inboundMesg.getAddress(),inboundMesg.getPort());
					
					System.out.println("PID: "+pid+" - received <"+inboundPaxosMesg.msgType+","+inboundPaxosMesg.cmdParamList+"> from PID: "+inboundPaxosMesg.senderId);
					protocolState = processClientReq(inboundPaxosMesg);
				}
				else if (inboundPaxosMesg.msgType.equals(PaxosConsts.PROMISE))
				{
					System.out.println("PID: "+pid+" - received <"+inboundPaxosMesg.msgType+" for instance: "+inboundPaxosMesg.instanceId+"> from PID: "+inboundPaxosMesg.senderId);
					instanceNumPromiseReceived.set(inboundPaxosMesg.instanceId,instanceNumPromiseReceived.get(inboundPaxosMesg.instanceId)+1);
					allInboundPromises.add(new String(inboundMesg.getData()));
					if (instanceNumPromiseReceived.get(inboundPaxosMesg.instanceId) == (PaxosConsts.NUM_PEERS/2)+1)
					{
						System.out.println("PID: "+pid+" - received majority of "+inboundPaxosMesg.msgType+" for instance: "+inboundPaxosMesg.instanceId);
						processPromise(allInboundPromises, inboundPaxosMesg.instanceId);
					}
					
				}
				else if (inboundPaxosMesg.msgType.equals(PaxosConsts.CHOSEN))
				{
					System.out.println("PID: "+pid+" - received <"+inboundPaxosMesg.msgType+" for instance: "+inboundPaxosMesg.instanceId+"> from PID: "+inboundPaxosMesg.senderId);
					instanceNumChosenReceived.set(inboundPaxosMesg.instanceId,instanceNumChosenReceived.get(inboundPaxosMesg.instanceId)+1);
					if (instanceNumChosenReceived.get(inboundPaxosMesg.instanceId) == (PaxosConsts.NUM_PEERS/2)+1)
					{
						System.out.println("PID: "+pid+" - received majority of "+inboundPaxosMesg.msgType+" for instance: "+inboundPaxosMesg.instanceId);
						
						protocolState = processChosen(inboundPaxosMesg,PaxosConsts.ALREADY_PROCESSED);
					}
					
				}
				else if (inboundPaxosMesg.msgType.equals(PaxosConsts.PREPARE_REJECT))
				{
					if (instancePrepareProposalNum.get(inboundPaxosMesg.instanceId) > inboundPaxosMesg.proposalNum)
					{
						System.out.println("PID: "+pid+" - IGNORING received message <"+inboundPaxosMesg.msgType+" for instance: "+inboundPaxosMesg.instanceId+"> from PID: "+inboundPaxosMesg.senderId+" with older prepare number");
						continue;
					}
					System.out.println("PID: "+pid+" - received <"+inboundPaxosMesg.msgType+" for instance: "+inboundPaxosMesg.instanceId+"> from PID: "+inboundPaxosMesg.senderId);
					processResendPrepare(inboundPaxosMesg);
				}
				else if (inboundPaxosMesg.msgType.equals(PaxosConsts.ALREADY_LEARNED))
				{
					System.out.println("PID: "+pid+" - received <"+inboundPaxosMesg.msgType+" for instance: "+inboundPaxosMesg.instanceId+" having value: "+inboundPaxosMesg.respToClient+"> from PID: "+inboundPaxosMesg.senderId);
					instanceProposalVal.set(inboundPaxosMesg.instanceId,inboundPaxosMesg.respToClient);
					protocolState = processChosen(inboundPaxosMesg, PaxosConsts.ALREADY_LEARNED);

				}
				else if (inboundPaxosMesg.msgType.equals(PaxosConsts.ALREADY_PROCESSED))
				{
					
					System.out.println("PID: "+pid+" - received <"+inboundPaxosMesg.msgType+" for instance: "+inboundPaxosMesg.instanceId+" having value: "+inboundPaxosMesg.respToClient+"> from PID: "+inboundPaxosMesg.senderId);
					instanceProposalVal.set(inboundPaxosMesg.instanceId,inboundPaxosMesg.respToClient);
					if (!instanceLearnList.get(inboundPaxosMesg.instanceId).equals(PaxosConsts.NO_ELE_STR))
					{
						
						protocolState = processChosen(inboundPaxosMesg,PaxosConsts.ALREADY_LEARNED);
					}
					else
					{
						protocolState = processChosen(inboundPaxosMesg,PaxosConsts.ALREADY_PROCESSED);
					}

				}
			}
			catch (SocketTimeoutException e) 
			{
				// TODO: handle exception
				// more than half died
				
			}
			catch (IOException e) 
			{
				// TODO: handle exception
			}

		}
		
	    return protocolState;
				
	}
	
	public int playClient()
	{
		String clientInp = null;
		ArrayList<String> cmdParamList = new ArrayList<String>();
		boolean continueRole = true;
		int clientCurrentLeader = 0;
		int numRespRec=0;
		while(continueRole)
		{
			System.out.print("\n Enter command(s): ");
			clientInp = new Scanner(System.in).nextLine();
			String cmdParam="";
			
			// Split multiple cmd string
			StringTokenizer clientInpTokens = new StringTokenizer(clientInp,";");
			while (clientInpTokens.hasMoreTokens())
			{
				// For each command get command (cmd) and the parameter (param)
				cmdParam = clientInpTokens.nextToken();
				cmdParamList.add(cmdParam);
			}
			
		
			byte clientOps[] = (new PaxosMessage()).buildPaxosMsg(PaxosConsts.CLIENT_REQ, pid, 
					PaxosConsts.NO_ELE_INT, PaxosConsts.NO_ELE_INT, PaxosConsts.NO_ELE_STR, cmdParamList, PaxosConsts.NO_ELE_INT,
					PaxosConsts.NO_ELE_STR,PaxosConsts.NO_ELE_STR,pid); 
			DatagramPacket clientPacket = null; 
			
			try
			{
				clientPacket = new DatagramPacket(clientOps, clientOps.length, InetAddress.getByName(remoteHost),upSet.get(clientCurrentLeader));
			}
			catch (UnknownHostException he) 
			{
				System.err.println(he.getMessage());
			}
			boolean leaderError = false;
			
			while(numRespRec < cmdParamList.size() ) 
			{
				try
				{
					pSocket.setSoTimeout(timeoutBound);
			 		if (!leaderError)
			 		{
						System.out.println("CID: "+pid+" - sending < "+cmdParamList.toString()+"> to PID:"+clientCurrentLeader);
			 			pSocket.send(clientPacket);
			 		}
			 		else
			 		{
			 			// send to all
			 			leaderError = false;
			 			System.out.println("CID: "+pid+" - LEADER with PID: "+clientCurrentLeader+" HAS CRASHED! Sending request to all processes.");
			 			byte resendClientOps[] = (new PaxosMessage()).buildPaxosMsg(PaxosConsts.CLIENT_REQ, pid, 
								PaxosConsts.NO_ELE_INT, PaxosConsts.NO_ELE_INT, PaxosConsts.NO_ELE_STR, cmdParamList, PaxosConsts.NO_ELE_INT,
								PaxosConsts.NO_ELE_STR,PaxosConsts.NO_ELE_STR,pid);
			 			DatagramPacket resendClientPacket = null;
			 			for (int i=0;i<upSet.size();i++)
			 			{
			 				if (upSet.get(i) != PaxosConsts.DEAD_PROCESS_PORT)
			 				{
				 				try
				 				{
				 					System.out.println("CID: "+pid+" - sending < "+cmdParamList.toString()+"> to PID:"+i);
				 					resendClientPacket = new DatagramPacket(resendClientOps, resendClientOps.length, 
				 							InetAddress.getByName(remoteHost),upSet.get(i));
				 					pSocket.send(resendClientPacket);
				 					
				 				}
				 				catch (UnknownHostException he) 
				 				{
				 					System.err.println(he.getMessage());
				 				}
			 				}
			 			}
			 			
			 		}
					DatagramPacket serverResp = new DatagramPacket(buf,buf.length);
					int clientRetries=0;
					int k=0;
					while(!leaderError && k < cmdParamList.size() )
					{
						try
						{
							if(!cmdParamList.get(k).equals(PaxosConsts.NO_ELE_STR))
							{
								Arrays.fill(serverResp.getData(), 0,PaxosConsts.MESG_SIZE,(byte)0);
								pSocket.receive(serverResp);
								clientRetries = 0;
								PaxosMessage dataFromServer = new PaxosMessage(serverResp.getData());
								int clientInstanceId = dataFromServer.instanceId;
								numRespRec++;
								if (cmdParamList.get(clientInstanceId).indexOf(PaxosConsts.EXIT) == -1)
								{
									cmdParamList.set(clientInstanceId, PaxosConsts.NO_ELE_STR);
								}
								else
								{
									continueRole = false;
								}
								System.out.println("CID: "+pid+" - received <"+dataFromServer.respToClient+"> from PID: "+dataFromServer.senderId);
								writeLog(dataFromServer.respToClient.trim().toString()+" from PID: "+dataFromServer.senderId);
								clientCurrentLeader = dataFromServer.senderId;
							}
							k++;
						}
						catch (SocketTimeoutException ste) 
						{
							System.err.println(ste.getMessage());
							clientRetries++;
							if (clientRetries < retriesBound)
							{
								try
								{
									pSocket.send(clientPacket);
								}
								catch (IOException io) {
									System.err.println(io.getMessage());
								}
							}
							else
							{
								leaderError = true;
								upSet.set(clientCurrentLeader, PaxosConsts.DEAD_PROCESS_PORT);
								
							}
						}
					}
				}
				catch(SocketException se)
				{
					System.err.println(se.getMessage());
				}
				catch(IOException io)
				{
					System.err.println(io.getMessage());
				}
			}
			if(cmdParamList.indexOf(PaxosConsts.EXIT) != -1)
			{
				System.out.println("CID: "+pid+" - Goodbye!");
				continueRole = false;
			}

		}
		return PaxosConsts.SESSION_DONE;
	}
	
	
	public void writeLog(String msg)
	{
		try
		{
			BufferedWriter bw = new BufferedWriter(new FileWriter(logFile,true));
			bw.write(msg+"\n");
			bw.flush();
		}
		catch(IOException io)
		{
			System.err.println(io.getMessage());
		}
	}
}

